#include "iPBNConnection.h"
#include "../util/h/iLogger.h"
#include "../util/src/iMsgThread.h"
#include "../util/src/iIOThread.h"
#include "../util/h/iMsgDB.h"

iPBNConnection::iPBNConnection(iSocket* so, iConnection* prvd, iAllocator* alloc)
:iOTCPConnection(so,prvd, alloc)
{
	m_conn_status = iConnection::CS_UNSPEC;
}
int32_t iPBNConnection::send_data(iMsgDB* data)
{
	FUNC("iPBNConnection::send_data");

	DEBUG("send_data, len="<<data->get_total_data_length());

	Packer pkr;
	pkr.pack(data->duplicate());
	return iOTCPConnection::send_data(pkr.package);
}
// worker thread
void iPBNConnection::on_connect_confirm()
{
	FUNC("iPBNConnection::on_connect_confirm");

	if(m_conn_status == CS_UNSPEC)
	{
		if(m_provider==NULL)
		{
			iMsgDB* mbs = m_data_q.front();
			if(mbs) 
			{
				DEBUG("send data="<<mbs<<",length="<<mbs->get_total_data_length());
				send_data(mbs);
			}
			m_data_q.pop_front();
			m_conn_status = CS_WAIT_REMOTE_RSP;
		}
		else
			m_conn_status = CS_WAIT_REQ;
	}
}
void iPBNConnection::on_data_indication(iMsgDB* mdb)
{
	FUNC("iPBNConnection::on_data_indication");

    INFO("PBNConnection::on_data_indication length="<<mdb->get_data_length());
	switch(m_conn_status)
	{
	case CS_ESTABLISHED:
		{
			iMsgDB* mbs = m_packer.extract(mdb);
			while(mbs)
            {
				m_jobber->post_msg(iMsg::new_instance(MSG_TP_DATA_IN, mbs, this), m_worker);
                mbs = m_packer.extract(NULL);
            }
		}
		break;
	case CS_WAIT_REQ:
		{
			iMsgDB* mbs = m_packer.extract(mdb);
            DEBUG("CS_WAIT_REQ: mbs="<<mbs<<",size:"<<(mbs ? mbs->get_total_data_length():0));
			if(mbs)
			{
				m_conn_status = CS_WAIT_LOCAL_RSP;
				struct payload {iTPConnection* conn; iMsgDB* request;} *pl = new payload;
				pl->conn = this;
				pl->request = mbs;
				m_jobber->post_msg(iMsg::new_instance(MSG_TP_CONNECTION_IN, pl, m_provider), m_worker);
			}
		}
		break;
	case CS_WAIT_REMOTE_RSP:
		{
			iMsgDB* mbs = m_packer.extract(mdb);
            iMsgID msg = MSG_TP_CONNECTION_IN;
			while(mbs)
			{
				m_conn_status = CS_ESTABLISHED;
				m_jobber->post_msg(iMsg::new_instance(msg, mbs, this), m_worker);
                msg = MSG_TP_DATA_IN;
                mbs = m_packer.extract(NULL);
			}
		}
		break;
	default:
		{
			DEBUG("unexpected data received,data="<<mdb<<",length=%d"<<mdb->get_total_data_length());
			mdb->release();
		}
		break;
	}
}
// jobber thread
int32_t iPBNConnection::connect_request(iMsgDB* request,  iTPConnection::iSink* sink)
{
	FUNC("iPBNConnection::connect_request");

	if(m_conn_status == CS_UNSPEC && m_sink==NULL)
	{
		m_sink = sink;
		m_worker->post_msg(iMsg::new_instance(MSG_TP_CONNECTION_OUT, request, this), m_jobber);
		return 0;
	}

	return -1;
}
int32_t iPBNConnection::connect_response(iMsgDB* response,iTPConnection::iSink* sink)
{
	FUNC("iPBNConnection::connect_response");

	if(m_conn_status == CS_WAIT_LOCAL_RSP)
	{
		m_sink = sink;
		if(response) send_data(response);
		return 0;
	}

	return -1;
}
int32_t iPBNConnection::get_option(TP_OPTION opt, void* optval, int32_t optlen)
{
    FUNC("iPBNConnection::get_option");

    switch(opt)
    {
    case TP_OPT_PROTO_TYPE:
        {
            *((int32_t*)optval) =  iTPConnection::PBN;
        }
        break;
    default:
        return iOTCPConnection::get_option(opt, optval, optlen);
    }

    return 0;
}
int iPBNConnection::on_message(iMsgID msg, void* param)
{
	switch(msg)
	{
	case MSG_TP_DATA_OUT:
		{
			if(m_conn_status == CS_WAIT_LOCAL_RSP)
				m_conn_status = CS_ESTABLISHED;

			return iOTCPConnection::on_message(msg, param);
		}
		break;
	case MSG_TP_CONNECTION_IN:
		{
			switch(m_conn_status)
			{
			case CS_ESTABLISHED:
				{
					m_sink->on_connect_confirm(0, (iMsgDB*)param);
				}
				break;
			default:
				break;
			}
		}
		return 0;
	default:
		return iOTCPConnection::on_message(msg, param);
	}
}
